package com.wutoon.etl.aop;

import com.alibaba.fastjson.JSON;
import com.wutoon.etl.annotation.PowerJobSSE;
import com.wutoon.etl.enums.SseStage;
import com.wutoon.etl.pojo.PowerJobResult;
import com.wutoon.etl.pojo.SseOutputInfo;
import com.wutoon.etl.util.PowerJobThreadPool;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicBoolean;


/**
 * powerJob sse接口拦截
 *
 * @author 武兴云/72176468
 * @date 2024/1/4 15:15
 * @update_by 武兴云/72176468
 * @update_at 2024/1/4 15:15
 * @creed if you have doubt , please contact me !!!
 */
@Slf4j
@Aspect
@Component
public class PowerJobSSEAspect {

    /**
     * 切点
     *
     * @param powerJobSSE powerJobSSE
     * @author 武兴云/72176468
     * @date 2024/1/4 14:20
     * @update_by 武兴云/72176468
     * @update_at 2024/1/4 14:20
     * @creed if you have doubt , please contact me !!!
     */
    @Pointcut("@annotation(powerJobSSE)")
    public void pointCut(PowerJobSSE powerJobSSE) {

    }

    /**
     * 环绕处理, 支持流式输出和json输出。取决于调用方是否带accept请求头为TEXT_EVENT_STREAM_VALUE
     *
     * @param joinPoint ProceedingJoinPoint
     * @param powerJobSSE PowerJobSSE
     * @return Object
     * @author 武兴云/72176468
     * @date 2024/1/4 14:25
     * @update_by 武兴云/72176468
     * @update_at 2024/1/4 14:25
     * @creed if you have doubt , please contact me !!!
     */
    @Around(value = "pointCut(powerJobSSE)", argNames = "joinPoint,powerJobSSE")
    public Object around(ProceedingJoinPoint joinPoint, PowerJobSSE powerJobSSE) throws Throwable {
        log.info("powerJobSSE interface start");
        HttpServletRequest request = ((ServletRequestAttributes) Objects.requireNonNull(RequestContextHolder.getRequestAttributes())).getRequest();
        HttpServletResponse response = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getResponse();

        String accept = request.getHeader("Accept");
        assert response != null;
        AtomicBoolean end = new AtomicBoolean(false);
        long interfaceStart = System.currentTimeMillis();
        if (StringUtils.equals(accept, MediaType.TEXT_EVENT_STREAM_VALUE)) {
            log.info("powerJobSSE interface is fluent");
            response.setContentType(MediaType.TEXT_EVENT_STREAM_VALUE);
            response.setCharacterEncoding("UTF-8");
            SseOutputInfo stage = new SseOutputInfo(SseStage.START.getCode(), "success");
            writeLine(response, JSON.toJSONString(stage));
            PowerJobThreadPool.STREAM_LOG_POOL.execute(() -> {
                // 开启异步定时心跳，否则nginx的proxy_read_timeout配置过小会导致连接断开
                // proxy_read_timeout配置含义：连续xx时长内需要收到至少一个字节才会认为连接存活
                while (!end.get() && System.currentTimeMillis() - interfaceStart < 3600000) {
                    SseOutputInfo info = new SseOutputInfo(SseStage.REPORT.getCode(), "heart beat, STREAM_LOG_POOL size:" + PowerJobThreadPool.STREAM_LOG_POOL.getActiveCount());
                    try {
                        Thread.sleep(30000);
                        if (!end.get()) {
                            writeLine(response, JSON.toJSONString(info));
                        }
                    } catch (Exception e) {
                        log.info("powerJobSSE aop stream heart beat sleep error");
                    }
                }
            });
        }
        try {
            Object result = joinPoint.proceed();
            if (StringUtils.equals(accept, MediaType.TEXT_EVENT_STREAM_VALUE)) {
                SseOutputInfo stage = new SseOutputInfo(SseStage.PROCESSING.getCode(), JSON.toJSONString(result));
                writeLine(response, JSON.toJSONString(stage));
                // 流式输出无需返回值，否则会报异常：没有合适的HttpMessageConverter
                return null;
            }
            log.info("powerJobSSE interface end!");
            return result;
        } catch (Exception e) {
            log.error("powerJobSSE tips: controller has uncaught exception:", e);
            if (StringUtils.equals(accept, MediaType.TEXT_EVENT_STREAM_VALUE)) {
                // 通知任务失败，否则任务会缺失失败信息
                SseOutputInfo stage = new SseOutputInfo(SseStage.PROCESSING.getCode(), JSON.toJSONString(new PowerJobResult<>(false, e.getMessage())));
                writeLine(response, JSON.toJSONString(stage));
                // 流式输出无需返回值，否则会报异常：没有合适的HttpMessageConverter
            }
            throw e;
        } finally {
            end.set(true);
        }
    }

    private static void writeLine(HttpServletResponse response, String data) throws IOException {
        response.getWriter().println("data: " + data);
        response.getWriter().println();
        response.getWriter().flush();
    }

}
